package com.gzhryc.common.kafka;

import com.gzhryc.common.kafka.procuder.MultipleKafkaProducer;
import com.gzhryc.common.kafka.procuder.SingleKafkaProducer;

import java.util.Properties;

public class KafkaProducerFactory {

	private static KafkaProducerFactory me = null ;

    private Properties properties;
    private JFKafkaProducer producer;

    private KafkaProducerFactory(){}

    public static KafkaProducerFactory self(){
		if(me == null) {
        	me = new KafkaProducerFactory() ;
        }
        return me ;
    }
    
    public KafkaProducerFactory initial(Properties properties) {
    	this.properties = properties;
    	String producerType = this.properties.getProperty("kafka.producerType","single");
    	if("multiple".equals(producerType)) {
    		String producerThreadNum = this.properties.getProperty("kafka.producerThreadNum","2");
    		producer = new MultipleKafkaProducer(properties,Integer.valueOf(producerThreadNum));
    	}else {
    		producer = new SingleKafkaProducer(properties);
    	}
    	return this;
    }
    
    public JFKafkaProducer getProducer() {
    	return producer;
    }
}
